package com.ruoyi.jihuo.controller;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ruoyi.common.constant.CacheConstants;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.jihuo.domain.JhTmProductsOnSale;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@RestController
@RequestMapping("/sse")
public class SseController {
    
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private RedisCache redisCache;
    @Autowired
    private ObjectMapper objectMapper;
    
    // 存储所有连接的SSE发射器
    private static final Map<String, SseEmitter> EMITTER_MAP = new ConcurrentHashMap<>();
    
    // 用于停止监听的标志 true停止 false继续
    private static final Map<String, Boolean> STOP_FLAGS = new ConcurrentHashMap<>();

    /**
     * 获取处理明细
     * @param taskId
     * @return
     */
    public List<JhTmProductsOnSale> getTaskInfoByRedis(String taskId) {
        // 从Redis获取任务信息
        String redisKey = CacheConstants.TASK_INFO + "-" + taskId;
        List<JhTmProductsOnSale> resultList = new ArrayList<>();

        List<String> cacheList = redisCache.getCacheList(redisKey);
        cacheList.forEach(item -> {
            try {
                JhTmProductsOnSale taskInfo = objectMapper.readValue(item, JhTmProductsOnSale.class);
                resultList.add(taskInfo);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        return resultList;
    }

    /**
     * 建立SSE连接并开始推送任务进度
     */
    @GetMapping("/task-progress")
    public SseEmitter streamTaskProgress(@RequestParam String taskId) {
        // 创建SSE发射器，设置超时时间(0表示永不超时)
        SseEmitter emitter = new SseEmitter(0L);
        String connectionId = taskId + "_" + System.currentTimeMillis();

        EMITTER_MAP.put(connectionId, emitter);
        STOP_FLAGS.put(connectionId, false);

        // 发送初始连接成功消息
        try {
            emitter.send(SseEmitter.event()
                .name("connect")
                .data("Connection established for task: " + taskId));
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 启动后台线程推送任务进度
        startProgressPushing(connectionId, taskId, emitter);

        // 连接完成或超时时的处理
        emitter.onCompletion(() -> {
            STOP_FLAGS.put(connectionId, true);
            EMITTER_MAP.remove(connectionId);
            System.out.println("SSE connection completed: " + connectionId);
        });

        emitter.onTimeout(() -> {
            STOP_FLAGS.put(connectionId, true);
            EMITTER_MAP.remove(connectionId);
            System.out.println("SSE connection timeout: " + connectionId);
        });

        emitter.onError((ex) -> {
            STOP_FLAGS.put(connectionId, true);
            EMITTER_MAP.remove(connectionId);
            System.out.println("SSE connection error: " + ex.getMessage());
        });

        return emitter;
    }
    
    /**
     * 后台线程：定期从Redis获取任务进度并推送
     */
    private void startProgressPushing(String connectionId, String taskId, SseEmitter emitter) {
        new Thread(() -> {
            // 存储上一次发送的processedQuantity值
            Integer lastProcessedQuantity = null;

            try {
                while (!STOP_FLAGS.getOrDefault(connectionId, true)) {
                    // 已完成处理的数量
                    Integer processedQuantity = redisCache.getCacheMapValue(
                            CacheConstants.TASK_PROGRESS + "-" + taskId,
                            "processedQuantity");
                    // 总数
                    Integer total = redisCache.getCacheMapValue(
                            CacheConstants.TASK_PROGRESS + "-" + taskId,
                            "total");

                    if (processedQuantity != null && total != null) {
                        // 任务是否完成
                        if (isTaskCompleted(processedQuantity, total)) {
                            emitter.send(SseEmitter.event()
                                    .name("complete")
                                    .data(total + "/" + total));
                            break;
                        }

                        // 只有当processedQuantity变化时才发送
                        if (lastProcessedQuantity == null ||
                                !lastProcessedQuantity.equals(processedQuantity)) {
                            // 发送进度数据
                            emitter.send(SseEmitter.event()
                                    .name("progress")
                                    .data(processedQuantity + "/" + total));
                            // 更新上一次发送的值
                            lastProcessedQuantity = processedQuantity;
                        }
                    } else {
                        emitter.send(SseEmitter.event()
                                .name("empty")
                                .data(0 + "/" + 0));
                        break;
                    }

                    // 每2秒检查一次
                    Thread.sleep(2000);
                }
            } catch (Exception e) {
                // 发生异常时关闭连接
                try {
                    emitter.completeWithError(e);
                } catch (Exception ex) {
                    // 忽略
                }
            } finally {
                STOP_FLAGS.put(connectionId, true);
                EMITTER_MAP.remove(connectionId);
            }
        }).start();
    }

    
    /**
     * 判断任务是否完成
     */
    private boolean isTaskCompleted(Integer processedQuantity, Integer total) {
        if (processedQuantity >= total){
            return true;
        }
        return false;
    }
    
    /**
     * 手动停止SSE连接
     */
    @GetMapping("/stop")
    public String stopSseConnection(@RequestParam String taskId) {
        String connectionIdPrefix = taskId + "_";
        EMITTER_MAP.keySet().stream()
            .filter(key -> key.startsWith(connectionIdPrefix))
            .forEach(key -> STOP_FLAGS.put(key, true));
        return "Stopped SSE connections for task: " + taskId;
    }
}